package com.gin.cep;

import com.gin.cep.vo.OrderInfo;
import com.gin.cep.vo.OrderMessage;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;


/**
 * 在京东里面，一个订单创建之后，15分钟内如果没有支付，会发送一个提示信息给用户，
 * 如果15分钟内已经支付的，需要发一个提示信息给商家
 *
 * @author gin
 * @date 2021/4/27
 */
public class OrderStateTest {

    public static void main(String[] args) {
        //获取运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //设置为事件时间时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        OutputTag<OrderMessage> payTimeoutTag = new OutputTag<OrderMessage>("pay_timeout") {
        };
        DataStreamSource<String> stream = env.readTextFile(Class.class.getResource("/OrderLog.csv").getPath());
        SingleOutputStreamOperator<OrderInfo> streamTime = stream
                .map(new RichMapFunction<String, OrderInfo>() {
                    @Override
                    public OrderInfo map(String resource) throws Exception {
                        String[] split = resource.split(",");
                        return new OrderInfo(split[0], split[1], split[2], Long.parseLong(split[3]));
                    }
                })
                .assignTimestampsAndWatermarks(//指定水印延迟10秒
                        new BoundedOutOfOrdernessTimestampExtractor<OrderInfo>(Time.seconds(10)) {
                            //将数据中的时间字段提取出来，转成Long类型,不改变输入的数据样式
                            @Override
                            public long extractTimestamp(OrderInfo loginEvent) {
                                //毫秒为单位
                                return loginEvent.getActionTime() * 1000L;
                            }
                        });

        KeyedStream<OrderInfo, String> keyedStream = streamTime.keyBy(new KeySelector<OrderInfo, String>() {
            @Override
            public String getKey(OrderInfo orderInfo) throws Exception {
                return orderInfo.getOid();
            }
        });

        SingleOutputStreamOperator<OrderMessage> select = keyedStream.process(new ProcessFunction<OrderInfo, OrderMessage>() {

            //状态直接保存订单创建的数据
            ValueState<OrderInfo> createOrderState;

            //状态中还必须保存触发时间
            ValueState<Long> timeOutState;

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<OrderMessage> out) throws Exception {
                OrderInfo createOrder = createOrderState.value();
                if (createOrder != null) { //触发器 开始触发了，订单创建之后过了15分钟，需要给用户一个提示信息 ,用户的提示信息放入侧流中
                    ctx.output(payTimeoutTag, new OrderMessage(createOrder.oid, " 该订单在15分钟内没有支付 请尽快支付", createOrder.actionTime, 0));
                    //清理状态
                    createOrderState.clear();
                    timeOutState.clear();
                }
            }

            @Override
            public void open(Configuration parameters) throws Exception {
                //初始化状态
                createOrderState = getRuntimeContext().getState(new ValueStateDescriptor<OrderInfo>("create_order", OrderInfo.class));

                timeOutState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("time_out", Long.class));
            }

            @Override
            public void processElement(OrderInfo value, Context ctx, Collector<OrderMessage> out) throws Exception {
                //首先从状态中取得订单的创建数据
                OrderInfo createOrder = createOrderState.value();
                if (value.status.equals("create") && createOrder == null) {//刚刚开始创建好订单
                    createOrderState.update(value);//把刚刚创建的订单信息存入状态
                    Long ts = value.actionTime * 1000L + (15 * 60 * 1000); //提示信息的触发器  触发的时间
                    timeOutState.update(ts);
                    //开始注册触发器, 在指定时间点触发(EventTimeTimer, 基于事件时间进行计算的)
                    ctx.timerService().registerEventTimeTimer(ts);
                }

                if (value.status.equals("pay") && createOrder != null) { //当前订单创建了，并且支付了
                    //在判断一下支付是否超时
                    if (timeOutState.value() > value.actionTime * 1000L) { //支付没有超时
                        //先把触发器删除, 避免已支付订单再触发消息通知
                        ctx.timerService().deleteEventTimeTimer(timeOutState.value());
                        //生成提示信息给商家
                        OrderMessage om = new OrderMessage(value.oid, "订单正常支付，请尽快发货!", createOrder.actionTime, value.actionTime);
                        out.collect(om); //发送到主流
                        //清理状态
                        timeOutState.clear();
                        createOrderState.clear();
                    }
                }
            }

        });

        select.getSideOutput(payTimeoutTag).print("侧流");
        select.print("主流");

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
